package com.aotain.nyx.ipstat;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map.Entry;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.common.mongo.DataColumn;
import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.nyx.flink.MongoSinkObject;
import com.aotain.nyx.util.CommonUtil;
/**
 * IP统计入库
 * <功能详细描述>
 * 
 * @author  Turk
 * @version  [版本号, 2017年4月21日]
 * @see  [相关类/方法]
 * @since  [产品/模块版本]
 */
public class IPFlowStatImport implements FlatMapFunction<Tuple2<String, IPStatTuple>, MongoSinkObject>{

	private HashMap<String,Double> mapAttackMax = new HashMap<String,Double>();
	
	private String tableName = "";
	
	private String currentDate = "";
	
	public IPFlowStatImport(String TableName) {
		tableName = TableName;
	}
	
	/**
	 * serialVersionUID
	 */
	private static final long serialVersionUID = -4292238510169755751L;

	@Override
	public void flatMap(Tuple2<String, IPStatTuple> arg0,
			Collector<MongoSinkObject> arg1) throws Exception {
		// TODO Auto-generated method stub
		//Logger.getLogger(IPFlowStatImport.class).info("MongoDB import Start...");
		//入库
		String dip = arg0.f0;
//		MongoConnection conn = MongoConnection.getInstance();
//		List<List<DataColumn>> data = new ArrayList<List<DataColumn>>();
		List<DataColumn> saverow = new ArrayList<DataColumn>();
		DataColumn colSIP = new DataColumn();
		colSIP.setKey("DESTIP");
		colSIP.setValue(dip);
		saverow.add(colSIP);
		
		SimpleDateFormat sdfm = new SimpleDateFormat("mm");
		SimpleDateFormat sdfh = new SimpleDateFormat("yyyyMMddHH");
		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateStrM = sdfm.format(date);
		String dateStrH = sdfh.format(date);
		int nMin5 = (Integer.parseInt(dateStrM)/5)*5;
		int nMin30 = (Integer.parseInt(dateStrM)/30)*30;
		String sMin5 = String.format("%02d", nMin5);
		String sMin30 = String.format("%02d", nMin30);
		String dateStr5 = dateStrH+sMin5 +"00";
		String dateStr30 = dateStrH+sMin30 +"00";
		String dateDay = sdfd.format(date);
		
		if(!currentDate.equals(dateDay)) {
			//跨天清理MAP
			mapAttackMax.clear();
			currentDate = dateDay;
		}
		
		
		String reporttime = "";
		//时间
		if(tableName.contains("5MIN")) {
			reporttime = dateStr5;
		} else if (tableName.contains("30MIN")) {
			reporttime = dateStr30;
		} else if (tableName.contains("HOUR")) {
			reporttime = dateStrH;
		} else if(tableName.contains("DAY")) {
			reporttime = dateDay;
		} else {
			return;
		}
	
	
		DataColumn colReportTime = new DataColumn();
		colReportTime.setKey("REPORTTIME");
		colReportTime.setValue(Long.parseLong(reporttime));
		saverow.add(colReportTime);
		
		
		
		IPStatTuple col = arg0.f1;
		if(col == null)
			return;
		List<DataColumn> incrow = new ArrayList<DataColumn>();
		for(Entry<String,Long> entry : col.getValueMap().entrySet()) {
			DataColumn colFlow = new DataColumn();
			String key = entry.getKey();
			Long value = entry.getValue();
			String column = key; //获取统计列名  上下行流量 包
			colFlow.setKey(column);
			colFlow.setValue(value);
			incrow.add(colFlow);
		}
		
		if(tableName.contains("DAY")) {
			String mapKeyMax = String.format("%s_%s", dip , dateDay); //天+IP
			Double attackMax = mapAttackMax.get(mapKeyMax);
			if(attackMax == null) {
				attackMax = col.getAttackMax();
				mapAttackMax.put(mapKeyMax, attackMax);
			}
			
			if(attackMax < col.getAttackMax()) {
				attackMax = col.getAttackMax();
				mapAttackMax.put(mapKeyMax, attackMax);
			}
			
			DataColumn colAttackMax = new DataColumn();
			colAttackMax.setKey("ATTACKMAX");
			colAttackMax.setValue(attackMax);
			saverow.add(colAttackMax);
		}
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("REPORTTIME",Long.parseLong(reporttime)));
		filter.add(new DataColumn("DESTIP",dip));
		
		MongoImportTool.getInstance().UpdateRowData(tableName, filter, saverow);
		MongoImportTool.getInstance().IncRowData(tableName, filter, incrow);
		
		
		
		
		
//		MongoSinkObject sinkObj = new MongoSinkObject();
//		sinkObj.setTableName(tableName);
//		sinkObj.setDmlType(DML.INSERT);
//		sinkObj.setImportData(row);
////		arg1.collect(sinkObj);
//		data.add(row);
//		conn.InsertData("sds", tableName, data);
		
		
		
		//Logger.getLogger(IPFlowStatImport.class).info("MongoDB import Row:" + data.size());
	}

}
